﻿using Akka.Actor;
using Akka.Routing;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Threading.Tasks;

namespace Dapper.Sharding.CAP
{
    public class ServerCap
    {
        #region MyRegion

        private readonly ActorSystem actorSystem;

        private readonly IActorRef actor;

        internal readonly ConcurrentDictionary<string, ResponseMessage> dict = new ConcurrentDictionary<string, ResponseMessage>();

        private string DefaultActorPath { get; }

        public string GetActorPath(string ip = null, int port = 0)
        {
            if (string.IsNullOrEmpty(ip))
            {
                ip = actorSystem.Settings.Config.GetString("akka.remote.dot-netty.tcp.hostname");
            }

            var myport = port.ToString();
            if (myport == "0")
            {
                myport = actorSystem.Settings.Config.GetString("akka.remote.dot-netty.tcp.port");
            }
            return $"akka.tcp://{actorSystem.Name}@{ip}:{myport}/user/{actor.Path.Name}";
        }

        #endregion

        public ServerCap(ActorSystem actorSystem, string actorName, int pool = 100)
        {
            this.actorSystem = actorSystem;
            var props = Props.Create<ServerActor>(this).WithRouter(new RoundRobinPool(pool));
            actor = this.actorSystem.ActorOf(props, actorName);
            DefaultActorPath = GetActorPath();
        }

        internal Task AskAsync(string clientActor, CommitMessage msg)
        {
            return actorSystem.ActorSelection(clientActor).Ask<bool>(msg, TimeSpan.FromSeconds(2));
        }

        internal void Tell(string clientActor, CommitTellMessage msg)
        {
            actorSystem.ActorSelection(clientActor).Tell(msg);
        }

        public async Task<ResponseMessage> ExecuteAsync(Func<string, string, Task[]> func, int RetryCount = 0)
        {
            var token = ShardingFactory.NextObjectId();
            try
            {
                var taskArry = func(token, DefaultActorPath);
                dict.TryAdd(token, new ResponseMessage() { TotalCount = taskArry.Length, RetryCount = RetryCount });
                //等待所有任务完成，这里开始阻塞，除非所有任务都异常，或者所有任务都完成，否则一直阻塞  
                await Task.WhenAll(taskArry);
            }
            catch
            {

            }
            var ok = dict.TryGetValue(token, out var response);
            if (ok)
            {
                lock (response._locker)
                {
                    response.Complete = true;
                    dict.TryRemove(token, out var _);
                }
            }
            return response;
        }

    }
}
